package com.spring.demo.webflux.demo.frame.jdk8;

import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * allOf&anyOf
 * 这两个方法的入参是一个completableFuture组、allOf就是所有任务都完成时返回。但是是个Void的返回值。
 * anyOf是当入参的completableFuture组中有一个任务执行完毕就返回。返回结果是第一个完成的任务的结果。
 *
 * https://www.zhihu.com/question/49618581 背压
 * Back Pressure "背压"
 * 是一种现象：在数据流从上游生产者向下游消费者传输的过程中，上游生产速度大于下游消费者速度，导致下游的Buffer溢出，这种现象叫做BackPressure出现。
 * 这句话的重点是Buffer 溢出 ；
 *
 * Backpressure 指的是在 Buffer 有上限的系统中，Buffer 溢出的现象；它的应对措施只有一个：丢弃新事件。
 *
 * 编程中的 Backpressure 这个概念源自工程概念中的 Backpressure：在管道运输中，气流或液流由于管道突然变细、
 * 急弯等原因导致由某处出现了下游向上游的逆向压力，这种情况称作「back pressure」。这是一个很直观的词
 * ：向后的、往回的压力——back pressure。可是，国内的热力工程界对这个词的正式翻译是「背压」，把「back」翻译成了「背」，着实有点让人无力吐槽。
 *
 * @author wangzhilong
 * @date 2021/3/12
 */
public class CompletableFutureTest {


    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        newFuture();

//        isDaemon();

//        allOfOrAnyOf();

//        join();

//        whenXXX  在一个任务之后调用的方法

        whenOpt();
    }

    /**
     * whenComplete 用户很神奇
     *
     * @throws InterruptedException
     */
    private static void whenOpt() throws InterruptedException {
        final CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("this is first task：" + Thread.currentThread());
//            try {
//                Thread.sleep(10);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
            int a = 1 / 0;
            return "first";
        }).whenComplete((s,e) ->{
            // whenComplete 这个操作可能使用主线程阻塞 也可能使用supplyAsync操作使用的线程
            System.out.println("whenComplete "+Thread.currentThread());
            try {
                Thread.sleep(100);
            } catch (InterruptedException interruptedException) {
                interruptedException.printStackTrace();
            }
            System.out.println("正确完成的结果：" + s+",错误信息："+e);
        });
        System.out.println("test");

        Thread.sleep(1000);
    }

    private static void join() {
        CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
                System.out.println("one test");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "result";
        }).join(); //阻塞主线程
        System.out.println("test");
    }

    private static void allOfOrAnyOf() throws InterruptedException, ExecutionException {
        //allOf就是所有任务都完成时返回。但是是个Void的返回值。
        //anyOf是当入参的completableFuture组中有一个任务执行完毕就返回。返回结果是第一个完成的任务的结果。
        final CompletableFuture<Object> allOf = CompletableFuture.anyOf(
                CompletableFuture.supplyAsync(() -> {
                    try {
                        Thread.sleep(3000);
                        System.out.println("future one over");
                    } catch (InterruptedException e) {
                        System.out.println("future one InterruptedException");
                    }
                    return "future one";
                }),
                CompletableFuture.supplyAsync(() -> {
                    try {
                        Thread.sleep(6000);
                        System.out.println("future two over");
                    } catch (InterruptedException e) {
                        System.out.println("future two InterruptedException");
                    }
                    return "future two";
                })
        );
        System.out.println("结果："+allOf.get());
    }

    /**
     * 这些任务中。带有supply是持有返回值的，run是void返回值的。
     * 在玩supply时发现一个问题如果使用supplyAsync任务时不使用任务的返回值。
     * 即不用get方法阻塞主线程会导致任务执行中断。(注：跟get方法无关，后面有答案)
     *
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void isDaemon() throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        executorService.submit(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                System.out.println("executorService 是否为守护线程：" + Thread.currentThread().isDaemon()+"," + Thread.currentThread().getName());
                return null;
            }
        });

        final CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> {
            System.out.println("supplyAsync 是否为守护线程：" + Thread.currentThread().isDaemon()+"," + Thread.currentThread().getName());
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //supplyAsync.get() 没有此句时任务下面的语句不会输出
            //原因是 CompletableFuture使用异步任务时的操作都是创建成了守护线程，那么我们没有调用get方法不阻塞这个主线程的时候，
            // 主线程执行完毕，所有的线程任务就会导致一个问题，就是守护线程退出。主线程退出守护线程就会退出。
            System.out.println("this is supplyAsync");
            return "supplyAsync 1";
        });

        final CompletableFuture<String> supplyAsync2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("supplyAsync2 使用executorService是否为守护线程：" + Thread.currentThread().isDaemon()+"," + Thread.currentThread().getName());
            System.out.println("this is supplyAsync 2");
            return "supplyAsync 2";
        },executorService);

        //当下面有非守护线程时
        //下面证实守护线程（supplyAsync）在其他非守护线程全部退出的情况下不继续执行。
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("new Thread 是否守护线程：" + Thread.currentThread().isDaemon());
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        executorService.shutdown();
        //注释掉 下面词句时，即不需要线程执行返回值时，
//        System.out.println(supplyAsync.get());
//        System.out.println(supplyAsync2.get());
        System.out.println("主线程执行完成");
    }

    private static void newFuture() throws InterruptedException, ExecutionException {
        // 创建CompletableFuture直接new对象也成。一个completableFuture对象代表着一个任务。
        // 这个对象能跟这个任务产生联系。下面用的complete方法意思就是这个任务完成了需要返回的结果。然后用get();方法可以获取到。
        final CompletableFuture<String> completableFuture = new CompletableFuture<>();
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("CompletableFuture 可以监控这个任务的执行,"+Thread.currentThread().getName());
                completableFuture.complete("任务执行完成返回");
            }
        }).start();

        System.out.println(completableFuture.get()+","+Thread.currentThread().getName());
    }




}
